package com.ruoyi.flink.source;

import com.ruoyi.common.utils.file.FileUtils;
import com.ruoyi.flink.bean.WindowTxt;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;

import java.io.File;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;

public class MyTxtSource implements SourceFunction<WindowTxt> {

    private volatile boolean running = true;

    private long lateTime = 10 * 1000;

    @Override
    public void run(SourceContext<WindowTxt> ctx) throws Exception {
        List<String> strings = FileUtils.readLines(new File("Z:\\gitee\\study_imooc\\test-data\\flink\\widow\\data1.txt"), "utf-8");
        DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        while (running) {
            synchronized (ctx.getCheckpointLock()) {
                strings.forEach(e -> {
                    WindowTxt windowTxt = new WindowTxt();
                    String[] s = e.split(" ");
                    windowTxt.setKey(s[0]);
                    windowTxt.setDate(s[1]);
                    windowTxt.setTime(s[2]);
                    windowTxt.setValue(Integer.parseInt(s[3]));
                    windowTxt.setEventTime(LocalDateTime.parse(s[1] + " " + s[2], dateTimeFormatter));

                    ctx.collectWithTimestamp(windowTxt, windowTxt.getEventTimeStamp());
                    ctx.emitWatermark(new Watermark(windowTxt.getEventTimeStamp() - lateTime));
                });

            }
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}
